package com.iot.service.mqtt.service.subscripe;

import com.iot.service.mqtt.model.entity.TlSubClient;
import com.iot.service.mqtt.model.topic.TlTopicTrie;
import com.iot.service.mqtt.service.StoreStrategy;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.topic.ITopicService;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: hszhou
 * @Date: 2025/1/8 9:24
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
@RequiredArgsConstructor
public class LocalSubscriptionServiceImpl implements ISubscriptionService, ITopicService {

    private static final Logger logger = LoggerFactory.getLogger(LocalSubscriptionServiceImpl.class);

    public static final ConcurrentHashMap<String, Set<TlSubClient>> TOPIC_TO_CLIENT = new ConcurrentHashMap<>();

    private final TlTopicTrie trie ;

    private final ISessionService sessionService;
    @Override
    public Mono<Boolean> subscribe(TlSubClient clientSub) {
        return Mono.fromSupplier(()->{
            String topic = clientSub.getTopic();
//           return TOPIC_TO_CLIENT.compute(topic,(k,v)->{
//                if(v==null){
//                    v= ConcurrentHashMap.newKeySet();
//                }
//                v.add(clientSub);
//                logger.debug("添加订阅关系【{}】",clientSub);
//                return v;
//            });
            trie.insert(topic,clientSub);
            return true;
        });
    }

    @Override
    public Mono<Boolean> unsubscribe(String clientId, String topic) {
        return Mono.fromSupplier(()->{
            TlSubClient subClient = new TlSubClient();
            subClient.setTopic(topic);
            subClient.setClientId(clientId);
//            TOPIC_TO_CLIENT.computeIfPresent(topic, (k, v) -> {
//                v.removeIf(next -> next.getClientId().equals(clientId));
//                return v.isEmpty() ? null : v; // 若 Set 为空，删除 topic 键
//            });
            trie.remove(topic,subClient);
            return true;
        });
    }

    @Override
    public Flux<TlSubClient> find(String topic) {
        return Flux.push(fluxSink-> {

//            TOPIC_TO_CLIENT.forEach((k, v)->{
//                if (match(topic,k)) {
//                    v.forEach(fluxSink::next);
//                }
//            }
            trie.search(topic).forEach(fluxSink::next);
        });
    }

    @Override
    public Mono<Boolean> clear(String clientId) {
        //获取这个客户端所有订阅的主题
            return sessionService.find(clientId)
            .flatMap(session -> Flux.fromIterable(session.getTopics())
                     .flatMap(topic -> unsubscribe(clientId, topic))
                     .then()
                     .thenReturn(true)
                     .onErrorReturn(false))
            .defaultIfEmpty(false);

    }
}
